package com.example.yckjbigdataservice.controller;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.UUID;
import java.util.concurrent.ExecutionException;

@RequestMapping("/kafka")
@RestController
@Slf4j
public class KafkaController {

    /*@Value("${custom.conf.kafka.producer-topic}")
    private String topic;*/

    private final KafkaProducer<String, String> kafkaProducer;

    public KafkaController(KafkaProducer<String, String> kafkaProducer) {
        this.kafkaProducer = kafkaProducer;
    }


    @GetMapping("/send/message")
    public String sendMessage(@RequestParam("topic") String topic, @RequestParam("message") String message) {
        try {
            kafkaProducer.send(new ProducerRecord<>(topic, UUID.randomUUID().toString(), message), new Callback() {
                /**
                 * 回调函数，用于处理异步发送模式下，消息发送到服务端后的处理。
                 *
                 * @param metadata  元数据信息
                 * @param exception 发送异常。如果没有错误发生则为Null。
                 */
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    long elapsedTime = System.currentTimeMillis();
                    if (metadata != null) {
                        log.info("message(" +  ", " + message + ") sent to partition(" + metadata.partition() + "), "
                                + "offset(" + metadata.offset() + ") in " + elapsedTime + " ms");
                    } else if (exception != null) {
                        log.error("The Exception occured.", exception);
                    }
                }
            }).get();
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException(e);
        }
        return "OK";
    }

}
